/**
 *	@file	work4.c
 *
 *	@brief	work thread which accept client, 
 *		and recv client request data, parse request data, 
 *		connect to server, send request to server and recv 
 *		reply from server, parse reply, and send reply to 
 *		client. 
 *	
 *	@author	Forrest.zhang	
 *
 *	@date	2008-07-31
 */

#define	_GNU_SOURCE

#include <unistd.h>
#include <signal.h>
#include <assert.h>
#include <stdlib.h>
#include <errno.h>
#include <string.h>
#include <stdio.h>
#include <sys/epoll.h>
#include <fcntl.h>

#include "debug.h"
#include "thread.h"
#include "proxy.h"
#include "session.h"
#include "ip_addr.h"
#include "sock_util.h"
#include "nb_splice.h"
#include "cpu_util.h"
#include "gcc_common.h"

//#define NDEBUG          1


static int _work_accept_client(int fd, int events, void *arg);
static int _work_check_server(int fd, int events, void *arg);
static int _work_recv_data(int fd, int events, void *arg);
static int _work_send_data(int fd, int events, void *arg);
static int _work_run_task(int fd, int task, void *arg);


/**
 *	Init work thread, alloc resources.
 *
 *	Return 0 if success, -1 on error.
 */
static int 
_work_init(thread_t *info)
{
	work_t *wi;

	assert(info);

	/* alloc work_info */
	wi = malloc(sizeof(work_t));
	if (!wi) {
		ERR("malloc error %s\n", ERRSTR);
		return -1;
	}
	memset(wi, 0, sizeof(work_t));

	wi->max = (g_proxy.max / g_proxy.nwork) + 1;
	
	/* alloc object pool */
	wi->pktpool = objpool_alloc(4096, 1024, 0);
	if (!wi->pktpool) {
		ERR("objpool_alloc failed\n");
		return -1;
	}
	DBG("work[%d] alloc packet pool(%p)\n", info->index, wi->pktpool);

	/* alloc session pool */
	wi->ssnpool = objpool_alloc(sizeof(session2_t), 1024, 0);
	if (!wi->ssnpool) {
		ERR("objpool_alloc failed\n");
		return -1;
	}
	DBG("work[%d] alloc session pool(%p)\n", info->index, wi->ssnpool);

	wi->fe = fd_epoll_alloc(g_proxy.maxfd, 1000);
	if (!wi->fe) {
		ERR("alloc fd_epoll failed\n");
		return -1;
	}
	DBG("work[%d] alloc fd_epoll(%p)\n", info->index, wi->fe);

	wi->httpfd = sk_tcp_server(&g_proxy.httpaddr, 1);
	if (wi->httpfd < 0) {
		ERR("create http listen fd failed\n");
		return -1;
	}
	sk_set_nonblock(wi->httpfd, 1);

	if (unlikely(fd_epoll_alloc_update(wi->fe, wi->httpfd, 
					   EPOLLIN, _work_accept_client)))
	{
		ERR("add http listen fd into fd epoll failed\n");
		return -1;
	}

	DBG("work[%d] create http listen fd %d\n", info->index, wi->httpfd);

	info->priv = wi;

	return 0;
}

/**
 *	Release work thread resource which alloced by _work_init()
 *
 *	No return.
 */
static void 
_work_free(thread_t *info)
{
	work_t *wi;

	assert(info);

	wi = info->priv;	
	if (!wi)
		return;

	if (wi->ssnpool)
		objpool_free(wi->ssnpool);
	
	if (wi->pktpool)
		objpool_free(wi->pktpool);

	if (wi->fe)
		fd_epoll_free(wi->fe);

	info->priv = NULL;
	free(wi);
}

/*
 *	Get a packet from pktpool if need, and put it to @q->pkt.
 *
 *	Return pointer to packet_t if success, NULL on error.
 */
static packet_t *
_work_get_packet(thread_t *info, session_t *s, session_queue_t *q)
{
	packet_t *pkt;
	work_t *wi;

	assert(info);
	assert(info->priv);
	assert(s);
	assert(q);

	wi = info->priv;

	pkt = q->pkt;
	if (pkt)
		return pkt;

	pkt = objpool_get(wi->pktpool);
	if (!pkt) {
		ERR("can't get packet from pkt pool\n");
		return NULL;
	}
	s->nalloced++;
	q->pkt = pkt;
	pkt->len = 0;
	pkt->sendpos = 0;
	pkt->pos = 0;
	pkt->capacity = wi->pktpool->objsize - sizeof(packet_t);

	if (q == &s->cliq) {
		FLOW(2, "work[%d]: ssn<%d> client %d get a packet(%p)\n", 
		     info->index, s->id, s->clifd, pkt);
	}
	else {
		FLOW(2, "work[%d]: ssn<%d> server %d get a packet(%p)\n",
		     info->index, s->id, s->svrfd, pkt);
	}

	return pkt;
}

/**
 *	Select a real server for session @s, use round-robin algo.
 *
 *	Return 0 if success, -1 on error.
 */
static int 
_work_select_server(thread_t *info, session_t *s)
{
	static int i = 0;
	work_t *wi;
	struct epoll_event e;
	int fd;
	int wait;
	int *pair;
	int index;

	assert(info);
	assert(info->priv);
	assert(s);

	wi = info->priv;

	if (s->svrfd > 0)
		return 0;

	proxy_stat_lock();
	index = i;
	i = (i + 1) % g_proxy.nrsaddr;
	proxy_stat_unlock();

	fd = sk_tcp_client_nb(&g_proxy.rsaddrs[index], &wait);
	if (fd < 0) {
		s->is_svrerror = 1;
		ERR("connect server failed\n");
		return -1;
	}

	sk_set_nodelay(fd, 1);
	sk_set_quickack(fd, 1);

	/* add it to epoll to wait connect success */
	if (wait) {

		/* add to epoll for write */
		pair = (int *)&e.data.u64;	
		pair[0] = fd;
		pair[1] = s->id;
		e.events = EPOLLOUT;			
		if (epoll_ctl(wi->epfd, EPOLL_CTL_ADD, fd, &e)) {
			s->is_svrerror = 1;
			close(fd);
			ERR("epoll add %d error %s\n", fd, ERRSTR);
			return -1;
		}
		s->is_svrep = 1;
		FLOW(3, "work[%d]: ssn<%d> server %d epoll add write\n",
		     info->index, s->id, fd);

		/* remove client from epoll */
		if (epoll_ctl(wi->epfd, EPOLL_CTL_DEL, s->clifd, NULL)) {
			s->is_clierror = 1;
			ERR("epoll del %d error %s\n", fd, ERRSTR);
			return -1;
		}
		s->is_cliep = 0;
		FLOW(3, "work[%d]: ssn<%d> client %d epoll del read\n",
		     info->index, s->id, s->clifd);

		s->is_svrwait = 1;
		wi->nblocked ++;
		FLOW(1, "work[%d]: ssn<%d> client %d connect to server %d wait\n", 
		     info->index, s->id, s->clifd, fd);
	}
	/* add to epoll if connect success */
	else {
		pair = (int *)&e.data.u64;	
		pair[0] = fd;
		pair[1] = s->id;
		e.events = EPOLLIN;			
		if (epoll_ctl(wi->epfd, EPOLL_CTL_ADD, fd, &e)) {
			ERR("epoll add %d error %s\n", fd, ERRSTR);
			s->is_svrerror = 1;
			close(fd);
			return -1;
		}
		s->is_svrep = 1;
		FLOW(3, "work[%d]: ssn<%d> server %d epoll add read\n",
		     info->index, s->id, fd);

		FLOW(1, "work[%d]: ssn<%d> client %d connect to server %d success\n", 
		     info->index, s->id, s->clifd, fd);
	}

	s->svrfd = fd;

	return 0;
}

static session2_t * 
_work_alloc_session(thread_t *info)
{
	work_t *wi;
	session2_t *s;

	assert(info);
	assert(info->priv);

	wi = info->priv;

	s = objpool_get(wi->ssnpool);
	if (unlikely(!s)) {
		ERR("objpool_get failed\n");
		return NULL;
	}
	memset(s, 0, sizeof(session2_t));
	s->cli_fd = -1;
	s->svr_fd = -1;

	return s;
}

/**
 *	Free session @s, release all resource used by @s.
 *
 *	No return.
 */
static void 
_work_free_session(thread_t *info, session2_t *s)
{
	int fd;

	assert(info);
	assert(s);

	/* close client socket */
	if (s->cli_fd > 0) {
		fd = s->cli_fd;
		
		FLOW(1, "work[%d]: ssn<%p> close client %d\n", 
		     info->index, s, fd);
		close(fd);
	}

	/* close server socket */
	if (s->svr_fd > 0) {
		fd = s->svr_fd;

		FLOW(1, "work[%d]: ssn<%p> close server %d\n", 
		     info->index, s, fd);
		close(fd);
	}

	/* delete session from session pool */
	FLOW(1, "work[%d]: ssn<%p> deleted\n", info->index, s);

	objpool_put(s);

	proxy_stat_lock();
	g_proxy.stat.nhttplive--;
	g_proxy.stat.nlive--;
	proxy_stat_unlock();
}

/**
 *	Accept new client and add to session table.
 *
 */
static int 
_work_accept_client(int fd, int events, void *arg)
{
	thread_t *info;
	work_t *wi;
	session2_t *s;
	fd_item_t *fi;
	int clifd;
	ip_port_t cliaddr;
	u_int64_t nlive;
	int i;
	char ipstr1[IP_STR_LEN];
	char ipstr2[IP_STR_LEN];

	assert(fd > 0);
	assert(arg);

	info = (thread_t *)arg;		
	assert(info->priv);

	wi = (work_t *)info->priv;

	if (unlikely(events & EPOLLERR))
		return -1;

	for (i = 0; i < 8; i++) {
		
		/* accept client */
		clifd = sk_tcp_accept(fd, &cliaddr);
		if (unlikely(clifd < 0)) {
			if (errno != EAGAIN) {
				ERR("accept %d failed: %s\n", fd, ERRSTR);
				return -1;
			}
			else {
				return 0;
			}
		}

		proxy_stat_lock();
		g_proxy.stat.naccept++;
		proxy_stat_unlock();
		
		sk_set_nonblock(clifd, 1);
		sk_set_nodelay(clifd, 1);
		sk_set_quickack(clifd, 1);

		/* check max connection */
		proxy_stat_lock();
		nlive = g_proxy.stat.nlive;
		proxy_stat_unlock();

		if (unlikely(nlive >= (u_int32_t)g_proxy.max)) {
			close(clifd);
			ERR("work[%d] too many client %lu max %d\n", 
			    info->index, nlive, g_proxy.max);
			return -1;
		}

		/* alloc new session */
		s = _work_alloc_session(info);
		if (unlikely(!s)) {
			ERR("alloc new session failed\n");
			close(clifd);
			return -1;
		}
		FLOW(1, "client %d accepted(%s->%s)\n", clifd,
                     ip_port_to_str(&cliaddr, ipstr1, IP_STR_LEN),
                     ip_port_to_str(&g_proxy.httpaddr, ipstr2, IP_STR_LEN));

                /* map fd to fd_item, and init it */
                fi = fd_epoll_map(wi->fe, clifd);
                assert(fi);
                fd_epoll_init_item(fi);
                fi->state = FD_READY;
                fi->priv = s;

                /* add new client fd to update events */
                if (unlikely(fd_epoll_alloc_update(wi->fe, clifd,
                                                   EPOLLIN | EPOLLRDHUP,
                                                   _work_recv_data)))
                {
                        ERR("%d alloc update failed\n", clifd);
                        goto out_failed;
                }
                FLOW(3, "client %d event alloc read|close\n", clifd);

                /* try to recv data in task after accept */
                if (unlikely(fd_epoll_alloc_task(wi->fe, clifd, FD_TASK_RECV, 
                                                 _work_run_task)) )
                {
                        ERR("%d alloc task failed\n", clifd);
                        goto out_failed;
                }
                FLOW(1, "client %d task alloc recv\n", clifd);  

                if (unlikely(_work_select_server(py, pl, s))) {
                        ERR("select pserver failed\n");
                        goto out_failed;
                }

		/* update proxy statistic */
		proxy_stat_lock();
		g_proxy.stat.nhttp++;
		g_proxy.stat.nhttplive++;
		g_proxy.stat.nlive++;
		proxy_stat_unlock();
	}

	return 0;
}


static int 
_work_check_server(thread_t *info, session2_t *s)
{
	work_t *wi;
	struct epoll_event event;
	int ret = 0;
	int fd;
	int *pair;

	assert(info);
	assert(info->priv);
	assert(s);

	wi = info->priv;
	fd = s->svrfd;

	ret = sk_is_connected(fd);
	if (ret <= 0) {
		FLOW(1, "work[%d]: ssn<%d> server %d connect failed\n", 
		     info->index, s->id, fd);
		return -1;
	}
	s->is_svrwait = 0;
	wi->nblocked --;

	/* mod it in epoll(read) */
	pair = (int *)&event.data.u64;
	pair[0] = fd;
	pair[1] = s->id;
	event.events = EPOLLIN;
	if (epoll_ctl(wi->epfd, EPOLL_CTL_MOD, fd, &event)) {
		ERR("epoll mod %d error %s\n", fd, ERRSTR);
		return -1;
	}
	FLOW(3, "work[%d]: ssn<%d> server %d epoll mod write->read\n",
	     info->index, s->id, fd);

	/* add client to epoll(read) */
	pair = (int *)&event.data.u64;
	pair[0] = s->clifd;
	pair[1] = s->id;
	event.events = EPOLLIN;
	if (epoll_ctl(wi->epfd, EPOLL_CTL_ADD, s->clifd, &event)) {
		ERR("epoll add %d error %s\n", s->clifd, ERRSTR);
		return -1;
	}
	FLOW(3, "work[%d]: ssn<%d> client %d epoll add read\n",
	     info->index, s->id, s->clifd);

	FLOW(1, "work[%d]: ssn<%d> client %d connect to server %d success\n", 
	     info->index, s->id, s->clifd, fd);

	return 0;
}


static int 
_work_recv_data(int fd, int events, void *arg)
{

}

static int 
_work_send_data(int fd, int events, void *arg)
{

}

static int 
_work_parse_data(thread_t *info, session2_t *s)
{

}

static int 
_work_run_task(int fd, void *arg)
{

}

/**
 *	work thread main loop
 *	
 *	Always return 0.
 */
static int 
_work_loop(thread_t *info)
{
	work_t *wi;
	
	assert(info);
	assert(info->priv);

	wi = (work_t *)info->priv;

	while (!g_proxy.stop) {
		fd_epoll_flush_update(wi->fe);
		fd_epoll_poll(info);
		fd_epoll_flush_task(wi->fe);
	}

	return 0;
}

/**
 *	The main function of work thread. 
 *
 *	Return NULL always.
 */
void *
work_run(void *arg)
{
	thread_t *info;
	int cpu;

	info = arg;
	assert(info);

	DBG("work[%d] started\n", info->index);

	/* bind to cpu */
	if (g_proxy.bind_cpu) {
		cpu = thread_bind_cpu(pthread_self(), info->index, 
				      g_proxy.bind_cpu_algo, 
				      g_proxy.bind_cpu_ht);
		if (cpu < 0) {
			ERR("work[%d] bind CPU failed", info->index);
			pthread_exit(0);
		}

		printf("work[%d] bind to cpu %d\n", info->index, cpu);
	}

	if (_work_init(info)) {
		if (g_proxy.main_tid > 0) {
			pthread_kill(g_proxy.main_tid, SIGINT);
		}
		pthread_exit(0);
	}

	_work_loop(info);

	_work_free(info);

	DBG("work[%d] stoped\n", info->index);

	pthread_exit(0);
}


